
package models.spider.actor;

import models.spider.ISpider;
import models.spider.request.Factory;
import models.spider.request.Request;
import models.spider.sink.ISink;
import models.spider.stats.Stopwatch;

import org.apache.commons.lang3.StringUtils;

import play.db.jpa.JPA;
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;

public class Spider extends UntypedActor implements ISpider {

    static class Message {

        protected String url = null;

        Message(final String url) {
            this.url = url;
        }

        public String getUrl() {
            return url;
        }
    }

    static class Crawl extends Message {

        Crawl(final String url) {
            super(url);
        }

        Crawl() {
            super(null);
        }

    }

    protected ActorRef       index;
    protected ActorRef       status;
    protected ISink          sink;
    protected Factory        requestFactory;
    protected LoggingAdapter log = null;

    public Spider(final ActorRef index, final ActorRef status, final models.spider.config.Factory config, Factory fact) {
        log = Logging.getLogger(getContext().system(), this);
        log.debug("spider new");

        this.index = index;
        this.status = status;
        this.sink = config.newSink();
        this.requestFactory = fact;
    }

    @Override
    public void onReceive(final Object message) throws Exception {
        // Date start = new Date();
        // log.debug("spider onReceive start");
        if (message instanceof Crawl) {
            crawl((Crawl) message);
        } else if (message instanceof Index.Next) {
            JPA.withTransaction(new play.libs.F.Callback0() {

                @Override
                public void invoke() throws Throwable {
                    next((Index.Next) message);
                }
            });
        } else {
            unhandled(message);
        }
        // Date end = new Date();
        // Long elapsed = (end.getTime() - start.getTime());
        // log.debug("spider onReceive done elapsed = " + elapsed);
    }

    protected void crawl(final Crawl message) {
        // log.debug("spider crawl");
        index.tell(new Index.Next(), getSelf());
        // log.debug("index path " + index.path().toString());
        // log.debug("spider crawl end");
    }

    protected void next(final Index.Next message) {
        Stopwatch total = new Stopwatch("Next all");
        // log.debug("spider next start ");
        // log.debug("spider next start thread id = " +
        // Thread.currentThread().getId());
        // log.debug("spider next start url=" + message.getUrl());
        String url = message.getUrl();
        if (StringUtils.isEmpty(url)) {
            return;
        }

        // List<String> links = new ArrayList<>();
        Request request = requestFactory.create(message.url);

        Stopwatch connect = new Stopwatch("Next connect");
        request.connect();
        connect.stop();
        log.debug("spider next connect elapsed =" + connect.elapsed());
        System.out.println("spider next connect elapsed =" + connect.elapsed());
        Stopwatch parse = new Stopwatch("Next parse");
        try {
            // must sink invalid request as well as valid one
            sink.sink(request, this);
        } catch (Exception ex) {
            log.debug("spider next error " + ex.getMessage());       
            System.out.println("spider next error " + ex.getMessage());
            //ex.printStackTrace();
        } finally {
            if (request.isValid()) {
                index.tell(new Index.Success(message.url), self());
            } else {
                index.tell(new Index.Failure(message.url), self());
            }
        }
        parse.stop();
        log.debug("spider parse elapsed =" + parse.elapsed());
        System.out.println("spider parse elapsed =" + parse.elapsed());
        index.tell(new Index.Next(), getSelf());

        total.stop();
        log.debug("spider next elapsed =" + total.elapsed());
        System.out.println("spider next elapsed =" + total.elapsed());

        // log.debug("spider next end url=" + message.getUrl());
        status.tell(new Status.Event(url, connect, parse, total), self());
    }

    @Override
    public void preStart() {
        // log.debug("spider pre start");
    }

    @Override
    public void postStop() {
        // log.debug("spider post stop");
    }

    @Override
    public void add(final String url) {
        index.tell(new Index.Add(url), self());
    }

    @Override
    public void refresh(final String url) {
        index.tell(new Index.Refresh(url), self());

    }

}
